package com.bytes32.stats;

import java.nio.DoubleBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Random;

public class MergeAndSelect {
	
	/*
	 * Determines the offset based on the size of the buffer the weight of the
	 * new collapsed buffer and the step in the iteration
	 */
	private static int offset(int wY, int j) {
		if (wY % 2 == 1)
			return (wY + 1) / 2;
		if (j % 2 == 0)
			return (wY) / 2;
		if (j % 2 == 1)
			return (wY + 2) / 2;
		throw new IllegalArgumentException(String.format(
				"Weight and J don't make sense wY:%d j:%d", wY, j));
	}
	
	/**
	 * Outputs the q-th order statistics
	 * @param buffers
	 * @param q
	 * @return
	 */
	static double[] naiveOutput(Collection<QBuffer> buffers, /* Quantiles */double... q){
		DoubleBuffer buffer = DoubleBuffer.allocate(q.length);
		/* collapse first */
		QBuffer result = collapse(buffers, 0); /* has no effect on the storage*/
		/* the quantiles will be printed in the requested order */
		for (double quantile : q){
			int index = Double.valueOf(Math.ceil(quantile * result.getMaxLength())).intValue();
			if (index < result.getMaxLength()) {
				buffer.put(result.get(index));
			} else {
				break; /* reached the end of the buffer, should print a warning here */
			}
		}
		return buffer.array();
	}
	/**
	 * Base strategy that computes the larges size of the
	 * buffers, total weight and total weight size
	 * 
	 * @author murariuf
	 *
	 */
	static abstract class MergeAndSelectStrategy{
		/**
		 * Compute the needed weight sums
		 * @param buffers
		 * @return size of the largest buffer
		 */
		/* Weight count for Y buffer */
		int wY = 0;
		
		/* sum over actual buffer size * buffer weight * */
		int totalWeightSize = 0;
		
		public int init(Collection<QBuffer> buffers) {
			int sizeK = -1;
			/* Sum the weights */
			for (QBuffer buf : buffers) {
				/* sum the buffer weights*/
				wY += buf.weight();
				/* Take the largest size of the buffer */
				sizeK = buf.getLength() > sizeK ? buf.getLength() : sizeK;
				
				totalWeightSize+=(buf.weight() * buf.getLength());
			}
			return sizeK;
		}
		/**
		 * returns the next step based on total weight
		 * and the current step
		 * @param j
		 * @param buffers
		 * @return
		 */
		abstract int next(/* step */ int j);
		/*
		 * Return the sum of the weight of all the buffers to be collapsed
		 */
		int getwY() {
			return wY;
		}
		
		/*
		 * Return the sum of the weights * size of the buffer
		 */
		int getTotalWeightSize() {
			return totalWeightSize;
		}
		
			
	}
	
	static QBuffer collapse(final Collection<QBuffer> buffers,
			final int newLevel){
		return mergeAndSelect(buffers, newLevel, new MergeAndSelectStrategy() {
			public int next(int j) {
				int offset = offset(wY, j);
				int pos = j * wY + offset;
				return pos;
			}
		});
	}
	
	static double[] output(final Collection<QBuffer> buffers, final double... q){
		InputBuffer result = mergeAndSelect(buffers, 0, new MergeAndSelectStrategy() {
			/* iterator over array of quantiles */
			DoubleBuffer nextq = DoubleBuffer.wrap(q);
			/* tracks changes in j, it is used to get the next position */
			int tracker = -1;
			
			int currentPos = -1;
			@Override
			int next(int j) {
				/* if j progresses and we still have quantiles */
				if (tracker != j){
					/* no more quantiles no more positions */
					if (!nextq.hasRemaining())
						return -1;
					double qth = nextq.get();
					if (qth <= 0 || qth >=1)
						throw new IllegalArgumentException("quantiles are in between (0,1) interval : unlike "+qth);
					currentPos = Double.valueOf(Math.ceil(totalWeightSize) * qth).intValue();
					tracker = j;
				}
				return currentPos;
			}
		});
		return result.toArray();
	}
	
	/**
	 * Merge two or more buffers
	 * 
	 * @param buffers
	 * @param sizeK
	 * @return
	 */
	private static InputBuffer mergeAndSelect(final Collection<QBuffer> buffers,
			final int newLevel, MergeAndSelectStrategy strategy) {
		int wY = 0;
		/* Weight count for Y buffer */
		int wCount = 0;
		/* counting indexes for every buffer to be merged */
		int[] index = new int[buffers.size()];
		/* Size of the new buffer */
		int sizeK = strategy.init(buffers);
		
		wY = strategy.getwY();
		/* If we have exhausted input buffers but not reached end position */
		boolean stop = true;

		/* Prepare the return buffer */
		double[] y = new double[sizeK];

		int j = 0;
		while (j < sizeK) {
			/* Expected position for this iteration */
			int pos = strategy.next(j);
			if (pos == -1)
				break;
			double min = Double.MAX_VALUE;
			int minBufId = 0;
			int bufId = 0;
			int minBufWeight = 0;
			for (QBuffer buf : buffers) {
				/* Look for min */
				if (index[bufId] >= buf.getLength()) {
					/* We're done with this buffer */
					bufId++;
					continue;
				}
				stop = false;
				double item = buf.get(index[bufId]);
				if (item < min) {
					min = item;
					minBufId = bufId;
					minBufWeight = buf.weight();
				}
				bufId++;
			}
			/*
			 * Increment the index of the buffer where the smallest item was
			 * picked from
			 */
			index[minBufId]++;
			/* Increment the weight and check if you pass the expected pos */
			if ((wCount+=minBufWeight) >= pos){
				y[j] = min;
				j++;
			}

			if (stop)
				break;
		}
		/* this happens for output */
		if (y.length!=j){
			y = Arrays.copyOf(y, j);
		}
		InputBuffer result = new InputBuffer(y, wY, newLevel, BufferStatus.FULL);
		return result;
	}
	
	/**
	 * Before reduce is called we need to have at most one full buffer and at most
	 * one partial buffer
	 * @param buffers
	 * @return
	 */
	public static Collection<QBuffer> preReduce(final Collection<QBuffer> buffers){
		Collection<QBuffer> finalList = new ArrayList<QBuffer>(2);
		/* at most one full buffer */
		QBuffer fullBuffer = null;
		/* last remaining buffer could be partial */
		QBuffer partialBuffer = null;
		Collection<QBuffer> fullBuffers = new ArrayList<QBuffer>();
		/* find the last buffer and the list of full buffers */
		for (QBuffer buffer : buffers){
			switch (buffer.status()) {
			case FULL:
				fullBuffers.add(buffer);
				break;
			case PARTIAL:
				partialBuffer = buffer;
				partialBuffer.sort();
				continue;
			case EMPTY:
				continue;
			}
		}
		if (partialBuffer != null){
			fullBuffer = collapse(fullBuffers, 0);
			finalList.add(fullBuffer);
			finalList.add(partialBuffer);
		}
		if (partialBuffer == null){
			fullBuffer = collapse(buffers,0);
			finalList.add(fullBuffer);
		}
		return finalList;
	}
	
	/**
	 * Samples values by sampleRate
	 * @param values
	 * @param sampleRate
	 * @return
	 */
	public static double[] sampleByRate(double[] values, int sampleRate){
		if (values == null || values.length == 0)
			throw new IllegalArgumentException("Array to sample cannot be null or empty");
		if (sampleRate < 1)
			throw new IllegalArgumentException("Sample rate cannot be smaller than 1");
		if (sampleRate == 1)
			return values.clone(); /* strange */
		DoubleBuffer wrapper = DoubleBuffer.wrap(values);
		int samplingSize = values.length / sampleRate;
		DoubleBuffer result = DoubleBuffer.allocate(samplingSize);
		Random rand = new Random();
		while (wrapper.hasRemaining()){
			double[] chunk = new double[Math.min(sampleRate, values.length-wrapper.position())];
			wrapper.get(chunk);
			result.put(chunk[rand.nextInt(chunk.length)]);
		}
		return result.array();
	}
		
}